/**
 * 
 */
package edu.umd.clip.lm.storage.remote;

import java.io.*;
import java.net.*;
import java.nio.channels.*;
import java.text.*;
import java.util.*;
import java.util.concurrent.atomic.AtomicLong;

import edu.umd.clip.lm.storage.*;
import edu.umd.clip.lm.util.*;

/**
 * @author Denis Filimonov <den@cs.umd.edu>
 *
 */
public class ClientStorageConnection {
	private SocketChannel channel;
	private SocketAddress address;
	private ConcurrentQueue<Message> msgToSend;
	private ClientListener listener;
	private AtomicLong bytesSent;
	private AtomicLong bytesReceived;
	private AtomicLong messagesSent;
	private AtomicLong messagesReceived;
	private AtomicLong responseProcessingTime = new AtomicLong();
	
	private Set<Integer> outstandingMessages;
	
	public ClientStorageConnection(SocketAddress address) throws IOException {
		this.address = address;
		this.connect();
		
		this.msgToSend = new ConcurrentQueue<Message>();
		this.listener = null;
		
		bytesSent = new AtomicLong();
		bytesReceived = new AtomicLong();
		messagesSent = new AtomicLong();
		messagesReceived = new AtomicLong();
		
		new SenderThread().start();
		new ReceiverThread().start();
		
		outstandingMessages = Collections.synchronizedSet(new HashSet<Integer>(50));
	}
	
	public synchronized void addListener(ClientListener storage) {
		listener = storage;
	}
	
	public void sendMessage(Message message) {
		doSendMessage(message);
		//msgToSend.offer(message);
	}

	private void doSendMessage(Message message) {
		while(true) {
			try {
				//System.err.println("Client Sender: sending a message");
				int size;
				synchronized(channel) {
					outstandingMessages.add(message.getId());
					size = message.send(channel);
					//System.out.printf("Sent message %d\n", message.getId());
				}
				messagesSent.incrementAndGet();
				bytesSent.addAndGet(size);
				//System.err.println("Client Sender: done sending a message");
				//channel.close();
				break;
			} catch (IOException e) {
				e.printStackTrace();
				reconnect();
			}
		}		
	}
	
	private class SenderThread extends Thread {
		private SenderThread() {
			super("client storage sender");
			setDaemon(true);
		}
		
		@Override
		public void run() {
			while(true) {
				Message msg = msgToSend.poll();
				doSendMessage(msg);
			}
		}
	}
	
	private class ReceiverThread extends Thread {
		private ReceiverThread() {
			super("client storage receiver");
			setDaemon(true);
		}

		@Override
		public void run() {
			while(!isInterrupted()) {
				try {
					Message message = Message.getMessage(channel);
					messagesReceived.incrementAndGet();
					bytesReceived.addAndGet(message.getSize());
					//System.out.printf("recevied a reply to %d\n", message.getReplyTo());
					outstandingMessages.remove(message.getReplyTo());
					int modelId = message.getModelId();
					try {
						long start = System.nanoTime();
						listener.receivedReply(message);
						long end = System.nanoTime();
						responseProcessingTime.addAndGet(end - start);
					} catch(ArrayIndexOutOfBoundsException e) {
						e.printStackTrace();
					} catch(NullPointerException e) {
						System.err.printf("non-registered storage: %d\n", modelId);
					}
				} catch (IOException e) {
					e.printStackTrace();
					reconnect();
				}
			}
		}
	}
	
	private boolean connect() throws IOException {
		channel = SocketChannel.open();
		channel.socket().setReceiveBufferSize(256*1024);
		channel.socket().setSendBufferSize(64*1024);
		channel.socket().setTcpNoDelay(true);
		channel.connect(address);
		return channel.finishConnect();
	}
	
	private synchronized void reconnect() {
		if (channel.socket().isConnected()) {
			try {
				channel.socket().close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		try {
			while(!Thread.currentThread().isInterrupted()) {
				try {
					System.err.printf("%s reconnecting to %s...", DateFormat.getDateTimeInstance().format(new Date()), address);
					if (!connect()) {
						System.err.println("Can't connect? Let's take a nap.");
						Thread.sleep(10000);
					} else {
						System.err.println("done");
					}
				} catch (IOException e1) {
					System.err.println("failed");
					e1.printStackTrace();
					Thread.sleep(60*1000);
				}
			}
			
			if (listener != null) {
				listener.reconnected();
			}
		} catch(InterruptedException e) {
			
		}
	}

	public long getBytesSent() {
		return bytesSent.get();
	}

	public long getBytesReceived() {
		return bytesReceived.get();
	}

	public long getMessagesSent() {
		return messagesSent.get();
	}

	public long getMessagesReceived() {
		return messagesReceived.get();
	}
	
	public long getResponseProcessingTime() {
		return responseProcessingTime.get();
	}
	
	public static class ClientMonitor extends TimerTask {
		final ClientStorageConnection connection;
		long lastReceivedHits = 0;
		long lastRequestedHits = 0;
		long lastLostRequests = 0;
		long lastTimeWaited = 0;
		long lastRequests = 0;
		long lastAsyncRequests = 0;
		long lastSyncTime = 0;
		long lastResponseDelayTime = 0;
		long lastSyncCount = 0;
		public static final long INTERVAL = 60000; 
		private NumberFormat nf;
		
		public ClientMonitor(ClientStorageConnection connection) {
			this.connection = connection;
			nf = new DecimalFormat("#.##");
		}


		/* (non-Javadoc)
		 * @see java.util.TimerTask#run()
		 */
		@Override
		public void run() {
			long receivedHits = ClientProbTreeStorage.getReceivedHits();
			long requestedHits = ClientProbTreeStorage.getRequestedHits();
			long lostRequests = ClientProbTreeStorage.getLostRequests();
			long timeWaited = ClientProbTreeStorage.getTimeWaited();
			long requests = connection.getMessagesReceived();
			long asyncRequests = ClientProbTreeStorage.getAsyncRequests();
			long syncTime = RequestBundle.getSyncTime();
			long responseDelayTime = RequestBundle.getResponseDelayTime();
			long syncCount = RequestBundle.getSyncCount();
			
			if (requests != lastRequests) {
				long diff = requests - lastRequests;
				System.err.printf("Client stats (1 min): %s req/s, %s async req/s, recvHits %s%%, reqHits %s%%, sync %sms avg, response delay time %sms avg\n", 
						nf.format((float) diff * 1000 / INTERVAL),
						nf.format((float) (asyncRequests - lastAsyncRequests) * 1000 / INTERVAL),
						nf.format((float) (receivedHits - lastReceivedHits) * 100 / diff),
						nf.format((float) (requestedHits - lastRequestedHits) * 100 / diff),
						//nf.format((float) (lostRequests - lastLostRequests) * 100 / diff),
						//nf.format((timeWaited - lastTimeWaited) / diff * 1e-6),
						nf.format((syncTime - lastSyncTime) / (syncCount - lastSyncCount) * 1e-6),
						nf.format((responseDelayTime - lastResponseDelayTime) / diff * 1e-6)
						);
			}
			lastReceivedHits = receivedHits;
			lastRequestedHits = requestedHits;
			lastLostRequests = lostRequests;
			lastTimeWaited = timeWaited;
			lastRequests = requests;
			lastAsyncRequests = asyncRequests;
			lastSyncTime = syncTime;
			lastResponseDelayTime = responseDelayTime;
			lastSyncCount = syncCount;
			//System.out.flush();
		} 
	}
}
